package com.pingan.haofang.searchcloud.batch.export;

import com.alibaba.druid.pool.DruidDataSource;
import com.google.common.collect.Lists;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.pingan.haofang.searchcloud.api.IndexRowData;
import com.pingan.haofang.searchcloud.batch.common.Batch;
import com.pingan.haofang.searchcloud.batch.common.BatchResult;
import com.pingan.haofang.searchcloud.batch.common.BatchRunner;
import com.pingan.haofang.searchcloud.index.dto.SqlResultDTO;
import net.sf.jsqlparser.JSQLParserException;
import net.sf.jsqlparser.expression.ExpressionVisitor;
import net.sf.jsqlparser.parser.CCJSqlParserUtil;
import net.sf.jsqlparser.schema.Table;
import net.sf.jsqlparser.statement.Statement;
import net.sf.jsqlparser.statement.select.Select;
import net.sf.jsqlparser.util.deparser.ExpressionDeParser;
import net.sf.jsqlparser.util.deparser.SelectDeParser;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.converter.json.GsonHttpMessageConverter;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.web.client.RestTemplate;

import java.io.BufferedReader;
import java.io.IOException;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * solrDataCompareWithMysql
 *
 * @author zhangbi617
 * @date 2017-07-11
 */
@Batch(value = "solrDataCompareWithMysql")
@ConditionalOnProperty("solrDataCompareWithMysql")
public class SolrDataCompareWithMysqlRunner implements BatchRunner, InitializingBean {

    private static final Logger LOGGER = LoggerFactory.getLogger(SolrDataCompareWithMysqlRunner.class);

    private JdbcTemplate exportJdbcTemplate;

    private RestTemplate restTemplate;

    /**
     * 指定数据库所在机器IP端口
     * -Dsolr.export.mysql.host=xxx
     */
    @Value("${solr.export.mysql.host:10.59.78.21}")
    private String exportMysqlHost;
    /**
     * 指定MySQL用户名
     * -Dsolr.export.mysql.username=xxx
     */
    @Value("${solr.export.mysql.username:root}")
    private String exportMysqlUserName;
    /**
     * 指定MySQL密码
     * -Dsolr.export.mysql.password=xxx
     */
    @Value("${solr.export.mysql.password:123456}")
    private String exportMysqlPassword;

    @Value("${search.cloud.search.url:http://10.59.78.25:9595/api/v1/sql/select}")

    private String searchCloudUrl;

    @Value("${search.cloud.search.url2:http://10.59.78.22:9595/api/v1/sql/select}")
    private String searchCloudUrl2;

    @Value("${search.cloud.test.sql.path:D:\\allsql.txt}")
    private String testSQLPath;

    Gson gson = new GsonBuilder().create();

    private static final String TYPE_SC1 = "sc1";
    private static final String TYPE_SC2 = "sc2";
    private static final String TYPE_MYSQL = "mysql";


    /**
     * 可以支持任意两个数据源比对
     * <p>
     * 例如 sc1-sc2,sc1-mysql,mysql-sc2,mysql-mysql
     *
     * @param args
     * @return
     */
    @Override
    public BatchResult run(String[] args) {

        String type = TYPE_SC1 + "-" + TYPE_SC2;
        if (args.length > 0) {
            type = args[0];
        }
        String[] types = type.split("-");
        String dsType1 = types[0];
        String dsType2 = types[1];

        try {
            try (BufferedReader br = Files.newBufferedReader(Paths.get(testSQLPath), StandardCharsets.UTF_8)) {
                String sql;
                while ((sql = br.readLine()) != null) {

                    // data1
                    List<Map<String, Object>> data1 = null;
                    try {
                        data1 = readFromDatasource(dsType1, sql);
                    } catch (JSQLParserException e) {
                        LOGGER.info("sql error {}", sql);
                    } catch (Exception e) {
                        LOGGER.error("read from {} error,{}", dsType1, sql);
                        continue;
                    }

                    // data2
                    List<Map<String, Object>> data2 = null;
                    try {
                        data2 = readFromDatasource(dsType2, sql);
                    } catch (JSQLParserException e) {
                        LOGGER.info("sql error {}", sql);
                    } catch (Exception e) {
                        LOGGER.error("read from {} error,{}", dsType2, sql);
                        continue;
                    }

                    // compare
                    try {
                        compare(data1, data2);
                        LOGGER.info("compare success,sql:{}", sql);
                    } catch (Exception e) {
                        LOGGER.warn("data compare fail,{} sql {}\n,solr:\n{}, db:\n{}", e.getMessage(), sql, gson.toJson(data1),
                                gson.toJson(data2));
                    }

                }
            }
        } catch (
                IOException e)

        {
            LOGGER.error(e.getMessage(), e);
            return BatchResult.FAIL;
        }
        return BatchResult.ALL_SUCCESS;
    }


    private List<Map<String, Object>> readFromDatasource(String type, String sql) throws JSQLParserException {
        switch (type) {
            case TYPE_SC1:
                return readFromSolr(searchCloudUrl, sql);
            case TYPE_SC2:
                return readFromSolr(searchCloudUrl2, sql);
            case TYPE_MYSQL:
                return readFromDB(sql);
            default:
                throw new RuntimeException("not support type");
        }
    }


    private void compare(List<Map<String, Object>> solrDatas, List<Map<String, Object>> dbDatas) {
        if (CollectionUtils.size(solrDatas) != CollectionUtils.size(dbDatas)) {
            throw new RuntimeException("data size not equal");
        }
        for (int i = 0; i < solrDatas.size(); i++) {
            Map<String, Object> solrRow = solrDatas.get(i);
            Map<String, Object> dbRow = dbDatas.get(i);
            if (!(solrRow.containsKey("suggestion") && solrRow.containsKey("_version_"))
                    && solrRow.size() != dbRow.size()) {
                Collection<String> solrMore = CollectionUtils.removeAll(solrRow.keySet(), dbRow.keySet());
                Collection<String> dbMore = CollectionUtils.removeAll(dbRow.keySet(), solrRow.keySet());

                throw new RuntimeException(i + ":data columns size is not equal,solrMore:" + solrMore + ",dbmore:" + dbMore);
            }

            for (Map.Entry<String, Object> entry : solrRow.entrySet()) {
                if (entry.getValue() != null && !entry.getValue().toString()
                        .equals(dbRow.get(entry.getKey()).toString())) {
                    throw new RuntimeException(i + ": " + entry.getKey() + " not equal");
                }
            }
        }
    }

    private List<Map<String, Object>> readFromDB(String sql) {
        exportJdbcTemplate.execute("use searchcloud_index_db;");
        List<Map<String, Object>> result = exportJdbcTemplate.queryForList(sql);
        return result;
    }

    /**
     * @param searchCloudUrl
     * @param sql
     * @return
     * @throws JSQLParserException
     */
    private List<Map<String, Object>> readFromSolr(String searchCloudUrl, String sql) throws JSQLParserException {

        SqlSelectForm form = new SqlSelectForm();
        form.sql = sql;

        form.indexName = parseIndexName(sql);

        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);
        HttpEntity request = new HttpEntity(form, headers);
        SqlResultDTO sqlResultDTO = restTemplate.postForObject(searchCloudUrl, request, SqlResultDTO.class);

        List<Map<String, Object>> result = new ArrayList<>();

        for (IndexRowData rowData : sqlResultDTO.getResult()) {
            Map<String, Object> map = new HashMap<>();
            for (Map.Entry<String, Object> entry : rowData.entrySet()) {
                map.put(entry.getKey(), entry.getValue());
            }
            result.add(map);
        }

        return result;
    }

    private String parseIndexName(String sql) throws JSQLParserException {
        Statement statement = CCJSqlParserUtil.parse(sql);
        Select select = (Select) statement;

        StringBuilder buffer = new StringBuilder();
        ExpressionDeParser expressionDeParser = new ExpressionDeParser();

        TableNameParser tableNameParser = new TableNameParser(expressionDeParser, buffer);

        expressionDeParser.setSelectVisitor(tableNameParser);
        expressionDeParser.setBuffer(buffer);
        select.getSelectBody().accept(tableNameParser);
        return tableNameParser.getTableName();
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        DruidDataSource druidDataSource = new DruidDataSource();
        druidDataSource.setUrl("jdbc:mysql://" + exportMysqlHost + "/searchcloud_index_db?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&rewriteBatchedStatements=false");
        druidDataSource.setUsername(exportMysqlUserName);
        druidDataSource.setPassword(exportMysqlPassword);
        druidDataSource.init();
        this.exportJdbcTemplate = new JdbcTemplate(druidDataSource);

        restTemplate = new RestTemplate();
        GsonHttpMessageConverter converter = new GsonHttpMessageConverter();
        converter.setDefaultCharset(Charset.forName("UTF-8"));
        restTemplate.setMessageConverters(Lists.newArrayList(converter));
    }

    private static class SqlSelectForm {
        private String indexName;
        private String sql;

        public String getSql() {
            return sql;
        }

        public void setSql(String sql) {
            this.sql = sql;
        }

        public String getIndexName() {
            return indexName;
        }

        public void setIndexName(String indexName) {
            this.indexName = indexName;
        }
    }

    public class TableNameParser extends SelectDeParser {
        private String tableName;


        public TableNameParser(ExpressionVisitor expressionVisitor, StringBuilder buffer) {
            super(expressionVisitor, buffer);
        }

        @Override
        public void visit(Table tableName) {
            this.tableName = tableName.getName();
        }

        public String getTableName() {
            return tableName;
        }

        public void setTableName(String tableName) {
            this.tableName = tableName;
        }
    }
}
